package com.youxin.dataStream.state;

import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

/**
 * flink引入了state和checkpoint才支持（at least once and exactly once）
 *      state一般指一个具体的task/operator的状态（这个数据是保存在java的堆内存中的）
 *      checkpoint可以理解为checkpoint是把state数据持久化存储了,表示了一个flink job在一个特定时刻的一份全局状态快照，即包含了task/operator的状态
 * 计算数据的状态(state taskManager不挂的话是可以恢复的)    flink中有两种基本类型的state
 *      keyed state
 *      operator state
 *      这两种基本类型又有两种形式存在， 原始状态（raw state ） 和 托管状态(managed state)
 *          托管状态是有flink框架管理的状态
 *          原始状态是由用户自行管理状态具体的数据结构，框架在做checkpoint的时候，使用byte[] 来读写状态内容
 */
public class KeyedStateText {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        DataStreamSource<Tuple2<Long, Long>> tuples = env.fromElements(Tuple2.of(1L, 3L), Tuple2.of(1L, 7L), Tuple2.of(1L, 4L), Tuple2.of(1L, 2L), Tuple2.of(2L, 3L),Tuple2.of(2L, 3L));
        tuples.keyBy(0)
                .flatMap(new CountWindowAverage())
                .print();
        env.execute("state");

    }


}
